package org.jxkj.util

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.jxkj.app.GetData
import org.jxkj.data.IDaoProvider

/**
 * 机组状态
 */
object JzCondition {

  def calData(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {

    /*
    import org.apache.spark.sql.types._
    val rdd = sc.parallelize(Seq(
      Row(1, "First Value"),
      Row(2, "Second Value")
    ))
    val df = ss.createDataFrame(rdd, schema)
   */

    import ss.implicits._
    val jzgkDF = Seq(
      ("JKXT_EJ_AI00440", "00F"),
      ("JKXT_EJ_AI00441", "01F"),
      ("JKXT_EJ_AI00442", "02F"),
      ("JKXT_EJ_AI00443", "03F"),
      ("JKXT_EJ_AI00444", "04F"),
      ("JKXT_EJ_AI00445", "05F"),
      ("JKXT_EJ_AI00446", "06F"),
      ("JKXT_EJ_AI00447", "07F"),
      ("JKXT_EJ_AI00448", "08F"),
      ("JKXT_EJ_AI00449", "09F"),
      ("JKXT_EJ_AI00450", "10F"),
      ("JKXT_EJ_AI00451", "11F"),
      ("JKXT_EJ_AI00452", "12F"),
      ("JKXT_EJ_AI00453", "13F"),
      ("JKXT_EJ_AI00454", "14F"),
      ("JKXT_EJ_AI00455", "15F"),
      ("JKXT_EJ_AI00456", "16F"),
      ("JKXT_EJ_AI00457", "17F"),
      ("JKXT_EJ_AI00458", "18F"),
      ("JKXT_EJ_AI00459", "19F"),
      ("JKXT_EJ_AI00460", "20F"),
      ("JKXT_EJ_AI00461", "21F")
    ).toDF("point_code", "unit_name")
    jzgkDF.createOrReplaceTempView("t_jzgk")
    jzgkDF.show()
    val arrayString: Array[String] = jzgkDF.map { x => x.getString(0) }.collect()

    //GetData.getBySecond(ss,sc,hTable,arrayString,time,step)
    GetData.getByWebApi(ss, sc, hTable, arrayString, time, step, 300)
    ss.sql("select a.unit_name as crew_code,b.createTime start_time,case when nvl(b.pvalue,1)=1 then '机组稳定运行态' else '非稳态' end working_name" +
      " from t_jzgk a join t_ods_hbase_edos_tmp b on a.point_code=b.checkPoint")
      .createOrReplaceTempView("t_jzgk_statu")
    //    val jzgkDF1=ss.sql("select * from t_jzgk_statu")
    //    jzgkDF1.filter("crew_code='18F'")show()

    ss.sql("select substr(start_time,1,10) start_time,count(distinct crew_code) unit_cnt" +
      " from t_jzgk_statu where working_name='机组稳定运行态'" +
      " and substr(start_time,12,8)='23:00:00' group by substr(start_time,1,10)").createOrReplaceTempView("t_jzgk_cnt")

    /*    val jzgkDF2=ss.sql("select * from t_jzgk_statu")
        jzgkDF2.filter("crew_code='18F'")show()*/
  }


  /**
   * 计算机组运行状态
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   * @param interval
   */
  def calData(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int, interval: Int): Unit = {
    import ss.implicits._
    // 机组工况测点
    val jzgkDF = Seq(
      ("JKXT_EJ_AI00440", "00F"),
      ("JKXT_EJ_AI00441", "01F"),
      ("JKXT_EJ_AI00442", "02F"),
      ("JKXT_EJ_AI00443", "03F"),
      ("JKXT_EJ_AI00444", "04F"),
      ("JKXT_EJ_AI00445", "05F"),
      ("JKXT_EJ_AI00446", "06F"),
      ("JKXT_EJ_AI00447", "07F"),
      ("JKXT_EJ_AI00448", "08F"),
      ("JKXT_EJ_AI00449", "09F"),
      ("JKXT_EJ_AI00450", "10F"),
      ("JKXT_EJ_AI00451", "11F"),
      ("JKXT_EJ_AI00452", "12F"),
      ("JKXT_EJ_AI00453", "13F"),
      ("JKXT_EJ_AI00454", "14F"),
      ("JKXT_EJ_AI00455", "15F"),
      ("JKXT_EJ_AI00456", "16F"),
      ("JKXT_EJ_AI00457", "17F"),
      ("JKXT_EJ_AI00458", "18F"),
      ("JKXT_EJ_AI00459", "19F"),
      ("JKXT_EJ_AI00460", "20F"),
      ("JKXT_EJ_AI00461", "21F")
    ).toDF("point_code", "unit_name")
    jzgkDF.createOrReplaceTempView("t_jzgk")
    val arrayString: Array[String] = jzgkDF.map { x => x.getString(0) }.collect()

    // 获取测点每一秒的值
    GetData.getBySecond(ss, sc, hTable, arrayString, time, step, interval)

    ss.sql(
      s"""
         | select
         |   a.unit_name as crew_code,
         |   b.createTime as start_time,
         |   (case when nvl(b.pvalue, 1) = 1 then '机组稳定运行态' else '非稳态' end) as working_name
         | from
         |   t_jzgk a join t_ods_hbase_edos_tmp b on
         |     a.point_code = b.checkPoint
         |""".stripMargin)
      .createOrReplaceTempView("t_jzgk_statu")

    ss.sql(
      s"""
         | select
         |   substr(start_time, 1, 10) as start_time,
         |   count(distinct crew_code) as unit_cnt
         | from
         |   t_jzgk_statu
         | where
         |   working_name = '机组稳定运行态'
         | and
         |   substr(start_time, 12, 8) = '23:00:00'
         | group by
         |   substr(start_time, 1, 10)
         |""".stripMargin)
      .createOrReplaceTempView("t_jzgk_cnt")
  }
}
